package com.atguigu.chapter07.b_watermarket;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @ClassName: Flink01_Tumbling_Window
 * @Description:
 * @Author: kele
 * @Date: 2021/4/6 20:06
 *
 *
 *
 **/
public class Flink10_Custom_WaterMarket {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //自定义水印
        WatermarkStrategy<WaterSensor> wms = new WatermarkStrategy<WaterSensor>() {
            @Override
            public WatermarkGenerator<WaterSensor> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {

                return new WatermarkGenerator<WaterSensor>() {

                    long maxTimestamp = Long.MIN_VALUE + 3000 + 1;  //时间间隔为3s

                    //每到达一个event，更新水印
                    // eventTimestamp:事件的时间戳
                    @Override
                    public void onEvent(WaterSensor event, long eventTimestamp, WatermarkOutput output) {

                        maxTimestamp = Math.max(maxTimestamp,eventTimestamp);

                        //设置按照事件发射水印，每到一个event，发射一个水印
                        output.emitWatermark(new Watermark(maxTimestamp - 3000 - 1));
                    }

                    //周期性的发射水印
                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {

                        //默认200ms发射一次水印
                        //output.emitWatermark(new Watermark(maxTimestamp - 3000 - 1));

                    }
                };
            }
        };



        env.socketTextStream("hadoop162",8888)
                .map(datas->{

                    String[] data = datas.split(",");
                    return new WaterSensor(data[0],
                            Long.parseLong(data[1]) * 1000,
                            Integer.valueOf(data[2]));
                })
                //设置分发水印
                .assignTimestampsAndWatermarks(
//                        设置自定义的水印
                        wms

                )
                .keyBy(WaterSensor::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context context,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {

                        int count = 0;
                        for (WaterSensor sensor : elements) {
                            count++;
                        }

                        String o = "key:"+key+
                                    ",["+context.window().getStart()/1000+
                                    ","+context.window().getEnd()/1000+")"+
                                    ",数据个数："+count;

                        out.collect(o);
                    }
                })
                .print();



        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
